package com.atguigu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DimUtil;
import com.atguigu.util.DruidPhoenixDSUtil;
import com.atguigu.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.util.ThreadUtil;

import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author hfh
 * @version 1.0
 * @date 2022/8/30 16:25
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T>{
    private ThreadPoolExecutor threadPoolExecutor=null;
    private DruidDataSource druidDataSource=null;
    private String tableName=null;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPoolExecutor= ThreadPoolUtil.threadPoolExecutor();
        druidDataSource= DruidPhoenixDSUtil.getDataSource();
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) {
        resultFuture.complete(Collections.singleton(input));
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture)  {
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                try{
                    //从druid连接池获取连接
                    DruidPooledConnection conn = druidDataSource.getConnection();
                    //拼接sql select * from table where id = key 读取对应维度表数据
                    String id = getKey(input);
                    JSONObject dimInfo = DimUtil.getDimInfo(conn, tableName, id);
                    //合并数据到input
                    if (dimInfo!=null){
                        join(input,dimInfo);
                    }

                } catch (SQLException e) {
                    e.printStackTrace();
                    System.out.println("关联维度表出错！");
                }

                resultFuture.complete(Collections.singleton(input));
            }
        });
    }
}
